package cn.tedu.micro.demo.client.demo01;

import cn.tedu.micro.demo.client.RabbitChannelManager;
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ConfirmListener;
import lombok.extern.slf4j.Slf4j;
import org.junit.jupiter.api.Test;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.sql.SQLOutput;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeoutException;

/**
 * 测试消息生产发送
 */
@Slf4j
public class Producer {
    //希望生产者发送的消息能够到达queue01
    @Test
    public void send01() throws IOException, InterruptedException {
        //1.组织一个消息对象 body headers routingKey
        String msg="hello rabbitmq,i am from 你好";
        //1.1消息体
        byte[] body=msg.getBytes(StandardCharsets.UTF_8);
        //1.2消息属性
        AMQP.BasicProperties.Builder build = new AMQP.BasicProperties.Builder();
        //build.userId("123");//消息发送的用户 admin 123456
        build.contentType("UTF-8");
        build.priority(1000);
        build.expiration(10000+"");//消息超时时间
        Map<String,Object> headers=new HashMap<>();
        headers.put("name","王翠花");
        headers.put("age",25);
        build.headers(headers);
        build.headers(headers);
        AMQP.BasicProperties props = build.build();
        //1.3 理论上说 路由key值可以自定义的 入门案例 路由key需要 是 队列名称
        String routingKey="queue01";
        //2.发送消息到 交换机
        Channel channel = RabbitChannelManager.newChannel();
        /*
            String exchange: 生产者只能将消息发送给交换机 必须在发送至指定交换机名字
         */
        channel.basicPublish("",routingKey,props,body);
        //Thread.sleep(10000000);
    }
    //发送确认--同步确认
    @Test
    public void sendSyncConfirm() throws IOException, InterruptedException, TimeoutException {
        Channel channel = RabbitChannelManager.newChannel();
        Long start=System.currentTimeMillis();
        channel.confirmSelect();//开启确认
        channel.basicPublish("","queue03",null,"hello".getBytes());
        boolean ok = channel.waitForConfirms(5000);//channel线程等待反馈 等不到就超时异常
        if (ok){
            System.out.println("发送成功");
        }else{
            System.out.println("发送失败");
        }
        long end=System.currentTimeMillis();
        System.out.println(end-start);
    }
    //异步确认
    @Test
    public void sendAsyncConfirm() throws IOException, InterruptedException {
        Channel channel = RabbitChannelManager.newChannel();
        Long start=System.currentTimeMillis();
        channel.confirmSelect();//开启确认
        //绑定监听对象 提供失败和成功对调方法
        channel.addConfirmListener(new ConfirmListener() {
            @Override
            public void handleAck(long deliveryTag, boolean multiple) throws IOException {
                System.out.println("接收到了rabbitmq反馈成功");
            }
            @Override
            public void handleNack(long deliveryTag, boolean multiple) throws IOException {
                System.out.println("接收到了rabbitmq反馈失败");
            }
        });
        channel.basicPublish("","queue03",null,"hello".getBytes());
        long end=System.currentTimeMillis();
        System.out.println(end-start);
        Thread.sleep(100000);
    }
    @Test
    public void sendBatchConfirm() throws IOException, InterruptedException, TimeoutException {
        Channel channel = RabbitChannelManager.newChannel();
        Long start=System.currentTimeMillis();
        channel.confirmSelect();
        //发送一条,或者发送多条.即使发送多条,循环执行,也不会发送,而是攒着.
        for (int i=0;i<10;i++){
            channel.basicPublish("","queue03",null,"hello".getBytes());
        }
        //统一确认结果
        try{
            //如果上述发送的某一条消息失败,确认结果就会抛出异常
            channel.waitForConfirmsOrDie(5000);
            //如果执行wait结束,说明所有消息全成功
        }catch (Exception e){
            log.error("某条消息失败",e);
        }
        long end=System.currentTimeMillis();
        System.out.println(end-start);

    }
}
